package com.hex.ds.hdtp.core.app.data.transfer.handler.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.hex.bigdata.hbda.utils.StringUtils;
import com.hex.ds.hdtp.core.app.common.entity.BasicContrastInfo;
import com.hex.ds.hdtp.core.app.common.util.JaveShellUtil;
import com.hex.ds.hdtp.core.app.data.contrast.service.IContrastTableStructureService;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferLogConverter;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferSqlStructureConverter;
import com.hex.ds.hdtp.core.app.data.transfer.dto.DataTransferSqlStructureDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.request.DataTransferLogRequestDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.response.DataTransferListResponseDto;
import com.hex.ds.hdtp.core.app.data.transfer.handler.IDataTransferHandlerJdbcService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferListService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferLogService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferTaskService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceConfInfoService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceInfoService;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcConnectUtil;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcUtil;
import com.hex.ds.hdtp.core.inf.data.contrast.service.impl.DataInfoSelectHandlerContext;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.repository.impl.SourceConfInfoRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.sql.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @Package com.hex.ds.hdtp.core.app.data.transfer.handler.impl
 * @ClassName AbstractDataTransferHandlerJdbcService
 * @Description 数据迁移JDBC方式抽象类
 * @Author Yin.Yang
 * @Date 2023/11/29 13:24
 * @Version v1.0
 **/
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public abstract class AbstractDataTransferHandlerJdbcService implements IDataTransferHandlerJdbcService {

    private final ISourceInfoService sourceInfoService;

    private final ISourceConfInfoService sourceConfInfoService;

    private final IDataTransferLogService dataTransferLogService;

    private final IDataTransferListService dataTransferListService;

    private final IDataTransferTaskService dataTransferTaskService;

    private final IDataTransferLogConverter dataTransferLogConverter;

    private final IDataTransferSqlStructureConverter dataTransferSqlStructureConverter;

    private final DataInfoSelectHandlerContext dataInfoSelectHandlerContext;

    private final IContrastTableStructureService contrastTableStructureService;

    private final SourceConfInfoRepository sourceConfInfoRepository;

    private static final String DELIMITER = "Separator";

    private static final String ENCODE = "Encode";

    private static final String BATCH_SIZE = "BatchSize";

    private static final String SHELL_RESULT = "SHELL_RESULT";

    private static final String SHELL_ERR_RESULT = "SHELL_ERR_RESULT";

    private static final String ANALYSIS_RESULT = "ANALYSIS_RESULT";

    private static final String SQL_WHERE = " WHERE ";

    /**
     * @Method dataJdbcTransfer
     * @param dataTransferSqlStructureDto
     * @param isDataAnalysis
     * @Return void
     * @Description 数据迁移执行
     * @Author Yin.Yang
     * @Date 2023/11/29 15:47
     * @Version V1.0
     */
    @Override
    @Async("dataTransferAsyncTaskExecutor")
    public void dataJdbcTransfer(DataTransferSqlStructureDto dataTransferSqlStructureDto, boolean isDataAnalysis) {
        boolean dataTransferSuccess = true;
        String dataAnalysisResult = dataAnalysis(dataTransferSqlStructureDto, isDataAnalysis);
        for (DateTime dateTime : dataTransferSqlStructureDto.getBizDates()) {
            String bizDate = DateUtil.format(dateTime, "yyyyMMdd");
            String dataTransferLogId = addDataTransferLog(dataTransferSqlStructureDto, bizDate, dataAnalysisResult, isDataAnalysis);
            if (!isDataAnalysis || StrUtil.isBlank(dataAnalysisResult)) {
                dataTransferSuccess = doDataJdbcTransfer(dataTransferLogId, bizDate, dataTransferSqlStructureDto);
            } else {
                dataTransferSuccess = false;
            }
        }
        updateDataTransferTask(dataTransferSqlStructureDto, dataTransferSuccess);
    }

    /**
     * @Method updateDataTransferLog
     * @param dataTransferLogPkId
     * @param resultMap
     * @Return void
     * @Description 更新数据迁移记录结果
     * @Author Yin.Yang
     * @Date 2023/11/29 15:43
     * @Version V1.0
     */
    private void updateDataTransferLog(String dataTransferLogPkId, Map<String, String> resultMap) {
        DataTransferLogRequestDto dataTransferLogRequestDto = new DataTransferLogRequestDto();
        dataTransferLogRequestDto.setPkId(dataTransferLogPkId).setEndTime(DateUtil.now());
        dataTransferLogRequestDto.setStatus("0".equals(resultMap.get(JaveShellUtil.SHELL_RESULT)) ? "SUCCESS" : "FAIL");
        dataTransferLogRequestDto.setAnalysisResult(resultMap.get(ANALYSIS_RESULT));
        String errorMessage = "转换执行SQL: " + resultMap.get("doTransferSql") + "\n" + "获取数据SQL: " + resultMap.get("querySourceSql") + "\n" + "迁移信息: " + resultMap.get(JaveShellUtil.SHELL_ERR_RESULT);
        dataTransferLogRequestDto.setErrorMessage(errorMessage);
        dataTransferLogService.modifyById(dataTransferLogRequestDto);
    }

    /**
     * @Method updateDataTransferTask
     * @param dataTransferSqlStructureDto
     * @param dataTransferSuccess
     * @Return void
     * @Description 更新数据迁移任务成功、失败数量和状态
     * @Author Yin.Yang
     * @Date 2023/11/29 15:43
     * @Version V1.0
     */
    private void updateDataTransferTask(DataTransferSqlStructureDto dataTransferSqlStructureDto, boolean dataTransferSuccess) {
        if (dataTransferSuccess) {
            dataTransferTaskService.modifySuccessNum(dataTransferSqlStructureDto.getDataTransferTaskPkId());
        } else {
            dataTransferTaskService.modifyFailNum(dataTransferSqlStructureDto.getDataTransferTaskPkId());
        }
        dataTransferTaskService.modifyStatus(dataTransferSqlStructureDto.getDataTransferTaskPkId());
    }

    /**
     * @Method dataAnalysis
     * @param dataTransferSqlStructureDto
     * @param isDataAnalysis
     * @Return java.lang.String
     * @Description 数据迁移前数据分析（源表和目标表是否存在、源表和目标表字段对比、源表是否为空）
     * @Author Yin.Yang
     * @Date 2023/11/29 15:43
     * @Version V1.0
     */
    private String dataAnalysis(DataTransferSqlStructureDto dataTransferSqlStructureDto, boolean isDataAnalysis) {
        if (isDataAnalysis) {
            SourceInfoPo sourceSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getSourceConnPkId());
            SourceInfoPo targetSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getTargetConnPkId());
            BasicContrastInfo basicContrastInfo = dataTransferSqlStructureConverter
                    .toBasicContrastInfo(dataTransferSqlStructureDto, sourceSource, targetSource);
            if (!contrastTableStructureService.contrastCreateTable(basicContrastInfo).getResult()) {
                return "源表或目标表不存在";
            } else if (!contrastTableStructureService.contrastFieldNameWithMap(basicContrastInfo, dataTransferSqlStructureDto.getDataTransferListPkId()).getResult()) {
                return "源表和目标表字段映射存在问题";
            } else if (!isSourceTableEmpty(sourceSource, dataTransferSqlStructureDto)) {
                return "源表数据为空，或源表数据量分析异常";
            }
        }
        return null;
    }

    /**
     * @Method isSourceTableEmpty
     * @param sourceSource
     * @param dataTransferSqlStructureDto
     * @Return boolean
     * @Description 源表数据是否为空判断
     * @Author Yin.Yang
     * @Date 2023/11/29 15:42
     * @Version V1.0
     */
    private boolean isSourceTableEmpty(SourceInfoPo sourceSource, DataTransferSqlStructureDto dataTransferSqlStructureDto) {
        try {
            Long sourceInfoNum = dataInfoSelectHandlerContext.getDataInfoService(sourceSource.getSourceType())
                    .selectIsHaveData(sourceSource.getSourceCode(), dataTransferSqlStructureDto.getSourceDatabase(),
                            dataTransferSqlStructureDto.getSourceTable());
            return sourceInfoNum > 0;
        } catch (Exception e) {
            log.error("数据迁移-数据分析-源表数据为空数据分析异常：{}", e.getMessage(), e);
            return false;
        }
    }

    /**
     * @Method addDataTransferLog
     * @param dataTransferSqlStructureDto
     * @param bizDate
     * @param dataAnalysisResult
     * @param isDataAnalysis
     * @Return java.lang.String
     * @Description 新增数据迁移记录
     * @Author Yin.Yang
     * @Date 2023/11/29 15:42
     * @Version V1.0
     */
    private String addDataTransferLog(DataTransferSqlStructureDto dataTransferSqlStructureDto, String bizDate,
                                      String dataAnalysisResult, boolean isDataAnalysis) {
        DataTransferListResponseDto dataTransferList = dataTransferListService.queryById(dataTransferSqlStructureDto.getDataTransferListPkId());
        DataTransferLogRequestDto dataTransferLog = dataTransferLogConverter.toDto(dataTransferList, bizDate, dataAnalysisResult);
        if (isDataAnalysis && StrUtil.isNotBlank(dataAnalysisResult)) {
            dataTransferLog.setStatus("FAIL").setEndTime(DateUtil.now());
        }
        return dataTransferLogService.add(dataTransferLog);
    }


    /**
     * @Method doDataJdbcTransfer
     * @param dataTransferLogId
     * @param bizDate
     * @param dataTransferSqlStructureDto
     * @Return boolean
     * @Description 分批处理源表数据
     * @Author Yin.Yang
     * @Date 2023/11/30 19:07
     * @Version V1.0
     */
    private boolean doDataJdbcTransfer(String dataTransferLogId, String bizDate, DataTransferSqlStructureDto dataTransferSqlStructureDto) {
        Map<String, String> resultMap = new HashMap();
        Map<String, String> sourceConfMap = dataTransferSqlStructureDto.getSourceConfMap();
        Map<String, String> targetConfMap = dataTransferSqlStructureDto.getTargetConfMap();
        String delimiter = getDelimiter(sourceConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType().toLowerCase() + DELIMITER));
        String targetEncoding = targetConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getTargetConfList()).getConfType().toLowerCase() + ENCODE);
        int targetBatchSize = Integer.parseInt(targetConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getTargetConfList()).getConfType().toLowerCase() + BATCH_SIZE));
        resultMap.put(SHELL_RESULT, "0");
        String querySourceSql = "SELECT * FROM " + dataTransferSqlStructureDto.getSourceTableName();
        String whereCondition = dataTransferSqlStructureDto.getWhereCondition();
        if (!StringUtils.isBlank(whereCondition)) {
            querySourceSql += SQL_WHERE + whereCondition;
        }
        try (ResultSet resultSet = executeQuerySql(dataTransferSqlStructureDto, resultMap, querySourceSql)) {
            ResultSetMetaData metaData = resultSet.getMetaData();
            int columnCount = metaData.getColumnCount();
            StringBuilder dataBuilder = new StringBuilder();
            int dataSize = 0;
            while (resultSet.next() && "0".equals(resultMap.get(SHELL_RESULT))) {
                dataSize++;
                processResultSet(resultSet, columnCount, dataBuilder, delimiter);
                // 每次读取到指定条数dataSize，写入目标数据库
                if (dataSize >= targetBatchSize) {
                    handleDataBuilder(dataTransferSqlStructureDto, dataBuilder, bizDate, delimiter, targetEncoding, resultMap);
                    dataBuilder = new StringBuilder();
                    dataSize = 0;
                }
            }
            if (dataSize > 0 && "0".equals(resultMap.get(SHELL_RESULT))) {
                handleDataBuilder(dataTransferSqlStructureDto, dataBuilder, bizDate, delimiter, targetEncoding, resultMap);
            }
        } catch (Exception e) {
            log.error("获取源表数据失败: {}", e.getMessage(), e);
            resultMap.put(SHELL_RESULT, "1");
            resultMap.put(SHELL_ERR_RESULT, e.getMessage() + e);
        } finally {
            updateDataTransferLog(dataTransferLogId, resultMap);
        }
        return "0".equals(resultMap.get(SHELL_RESULT));
    }

    /**
     * @Method executeQuerySql
     * @param dataTransferSqlStructureDto
     * @param resultMap
     * @Return java.sql.ResultSet
     * @Description 执行查询SQL获取源表数据
     * @Author Yin.Yang
     * @Date 2023/11/30 17:23
     * @Version V1.0
     */
    private ResultSet executeQuerySql(DataTransferSqlStructureDto dataTransferSqlStructureDto, Map<String, String> resultMap, String querySourceSql) throws SQLException {
        SourceInfoPo sourceSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getSourceConnPkId());
        List<SourceConfInfoPo> sourceConfInfoPoList = sourceConfInfoService.queryListBySourceCode(sourceSource.getSourceCode());

        resultMap.put("querySourceSql", querySourceSql);
        try(Connection jdbcConnection = JdbcUtil.getJdbcConnection(sourceConfInfoPoList);
            Statement jdbcStatement = JdbcUtil.createJdbcStatement(jdbcConnection);) {
            return jdbcStatement.executeQuery(querySourceSql);
        }

    }

    /**
     * @Method processResultSet
     * @param resultSet
     * @param columnCount
     * @param dataBuilder
     * @param delimiter
     * @Return void
     * @Description 处理结果集数据
     * @Author Yin.Yang
     * @Date 2023/11/30 17:23
     * @Version V1.0
     */
    private void processResultSet(ResultSet resultSet, int columnCount, StringBuilder dataBuilder, String delimiter) throws SQLException {
        for (int i = 1; i <= columnCount; i++) {
            String columnValue = resultSet.getString(i);
            if (columnValue == null) {
                dataBuilder.append("");
            } else if(columnValue.contains("\n") || columnValue.contains("\r") || columnValue.contains("\\")) {
                dataBuilder.append(columnValue
                        .replace("\n", "\\n")
                        .replace("\r", "\\r")
                        .replace("\\", "\\\\")
                        .trim()
                );
            } else {
                dataBuilder.append(columnValue.trim());
            }
            if (i < columnCount) {
                dataBuilder.append(delimiter);
            }
        }
        dataBuilder.append("\n");
    }

    /**
     * @Method handleDataBuilder
     * @param dataTransferSqlStructureDto
     * @param dataBuffer
     * @param bizDate
     * @param delimiter
     * @param targetEncoding
     * @param resultMap
     * @Return void
     * @Description 处理批数据后写入目标库
     * @Author Yin.Yang
     * @Date 2023/12/1 9:55
     * @Version V1.0
     */
    private void handleDataBuilder(DataTransferSqlStructureDto dataTransferSqlStructureDto, StringBuilder dataBuffer, String bizDate, String delimiter, String targetEncoding, Map<String, String> resultMap) {
        StringBuilder sourceData = addBufferPartition(dataBuffer, bizDate, delimiter);
        writeTarget(sourceData, dataTransferSqlStructureDto, delimiter, targetEncoding, resultMap);
    }

    /**
     * @Method writeTarget
     * @param sourceData
     * @param dataTransferSqlStructureDto
     * @param delimiter
     * @param targetEncoding
     * @param resultMap
     * @Return void
     * @Description 写数据到目标库
     * @Author Yin.Yang
     * @Date 2023/12/1 9:55
     * @Version V1.0
     */
    private void writeTarget(StringBuilder sourceData, DataTransferSqlStructureDto dataTransferSqlStructureDto, String delimiter, String targetEncoding, Map resultMap) {
        try {
            // 建立目标数据库连接
            SourceInfoPo targetSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getTargetConnPkId());
            List<SourceConfInfoPo> targetConfInfoPoList = sourceConfInfoService.queryListBySourceCode(targetSource.getSourceCode());
            // 提交源表中的数据
            submitSourceData(targetConfInfoPoList, sourceData, dataTransferSqlStructureDto.getTargetTableName(), delimiter, targetEncoding, resultMap);
            resultMap.put(SHELL_RESULT, "0");
            resultMap.put(SHELL_ERR_RESULT, "迁移成功");
            resultMap.put(ANALYSIS_RESULT, getResultCount(dataTransferSqlStructureDto));
        } catch (Exception e) {
            log.error("【{}】数据同步失败: {}", dataTransferSqlStructureDto.getDataTransferSqlFileName(), e.getMessage(), e);
            resultMap.put(SHELL_RESULT, "1");
            resultMap.put(ANALYSIS_RESULT, "迁移失败,无法汇总");
            resultMap.put(SHELL_ERR_RESULT, e.getMessage() + e);
        }
    }

    /**
     * @Method addBufferPartition
     * @param sourceData
     * @param bizDate
     * @param delimiter
     * @Return java.lang.StringBuffer
     * @Description 向源数据拼接分区字段
     * @Author Yin.Yang
     * @Date 2023/11/29 15:37
     * @Version V1.0
     */
    private StringBuilder addBufferPartition(StringBuilder sourceData, String bizDate, String delimiter) {
        Pattern pattern = Pattern.compile("\n");
        String resultString = pattern.splitAsStream(sourceData.toString())
                .map(line -> line + delimiter + bizDate)
                .collect(Collectors.joining("\n"));
        return new StringBuilder(resultString);
    }

    /**
     * @Method getDelimiter
     * @param delimiter
     * @Return java.lang.String
     * @Description 获取正确的分隔符
     * @Author Yin.Yang
     * @Date 2023/11/29 15:37
     * @Version V1.0
     */
    private String getDelimiter(String delimiter) {
        // 如果分隔符符合\00X则转为ascii码，否则直接使用
        if (delimiter.startsWith("\\") && delimiter.length() > 1) {
            int asciiCode = Integer.parseInt(delimiter.substring(3));
            delimiter = Character.toString((char) asciiCode);
        }
        return delimiter;
    }

    /**
     * @Method getResultCount
     * @param dataTransferSqlStructureDto
     * @Return java.lang.String
     * @Description 数据汇总情况
     * @Author Yin.Yang
     * @Date 2023/11/29 15:37
     * @Version V1.0
     */
    private String getResultCount(DataTransferSqlStructureDto dataTransferSqlStructureDto) {
        SourceInfoPo sourceSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getSourceConnPkId());
        SourceInfoPo targetSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getTargetConnPkId());
        List<SourceConfInfoPo> sourceConfInfoList = sourceConfInfoRepository.queryListBySourceCode(sourceSource.getSourceCode());
        List<SourceConfInfoPo> targetConfInfoList = sourceConfInfoRepository.queryListBySourceCode(targetSource.getSourceCode());
        String sql = "SELECT COUNT(*) FROM %s.%s";
        String sourceSql = String.format(sql, dataTransferSqlStructureDto.getSourceDatabase(), dataTransferSqlStructureDto.getSourceTable());
        String targetSql = String.format(sql, dataTransferSqlStructureDto.getTargetDatabase(), dataTransferSqlStructureDto.getTargetTable());
        if (StringUtils.isNotBlank(dataTransferSqlStructureDto.getWhereCondition())) {
            sourceSql = sourceSql + SQL_WHERE + dataTransferSqlStructureDto.getWhereCondition();
            targetSql = targetSql + SQL_WHERE + dataTransferSqlStructureDto.getWhereCondition();
        }
        Long sourceLongResult = null;
        Long targetLongResult = null;
        try {
            sourceLongResult = JdbcConnectUtil.getLongResult(sourceConfInfoList, sourceSql);
            targetLongResult = JdbcConnectUtil.getLongResult(targetConfInfoList, targetSql);
        } catch (SQLException e) {
            Throwable rootCause = findRootCause(e);
            if (rootCause != null) {
                return rootCause.getMessage();
            }
        }
        return CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType() + "数据库数据总量：" + sourceLongResult.toString() + "\n" + CollUtil.getFirst(dataTransferSqlStructureDto.getTargetConfList()).getConfType() + "数据库数据总量：" + targetLongResult.toString();
    }

    /**
     * @Method findRootCause
     * @param e
     * @Return java.lang.Throwable
     * @Description 辅助方法，递归查找异常根本原因
     * @Author Yin.Yang
     * @Date 2023/11/29 15:37
     * @Version V1.0
     */
    private static Throwable findRootCause(Throwable e) {
        Throwable cause = e;
        while (cause.getCause() != null) {
            cause = cause.getCause();
        }
        return cause;
    }

    /**
     * @Method writeTarget
     * @param sourceData
     * @param targetConfInfoPoList
     * @param delimiter
     * @param targetEncoding
     * @param resultMap
     * @Return void
     * @Description 具体写数据到目标数据库
     * @Author Yin.Yang
     * @Date 2023/11/29 15:38
     * @Version V1.0
     */
    protected abstract void submitSourceData(List<SourceConfInfoPo> targetConfInfoPoList, StringBuilder sourceData, String targetTableName, String delimiter, String targetEncoding, Map resultMap) throws SQLException, IOException;

}
